package com.atguigu.chapter07.d_state;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;

/**
 * @ClassName: Flink01_State_Operator_List
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 19:38
 *
 *
 * 键控状态之 ReduceState
 *
 *
 * 案例：计算每个传感器的水位和
 * 思路：获取reduce的状态，初始化定义状态的计算过程，在处理函数中直接将数据值放入到状态中
 *
 *
 * 1、先重写open方法，获取状态
 *      获取状态 reduceState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Integer>(
 *                                 "reduceState",   //reduce名称
 *                                 new ReduceFunction<Integer>() {   //处理函数
 *                                     @Override
 *                                     public Integer reduce(Integer value1, Integer value2) throws Exception {
 *                                         //计算过程......
 *                                     }
 *                                 },
 *                                 xxx.class
 *                                 )
 *
 * 2、数据处理更新状态
 *      将得到的值放入到状态中  reduceState.add(value)
 **/
public class Flink05_Key_State_Reduce {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        env.socketTextStream("hadoop162",8888)
                .map(datas->{

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    private ReducingState<Integer> reduceState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        /**
                         *
                         */
                        reduceState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Integer>("reduceState",
                               /* new ReduceFunction<Integer>() {
                                    @Override
                                    public Integer reduce(Integer value1, Integer value2) throws Exception {
                                        return value1 + value2;
                                    }
                                },*/
                               //简化，使用Lambud表达式
                               (value1, value2) -> value1 + value2,
                                Integer.class));
                    }

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        reduceState.add(value.getVc());

                        out.collect(ctx.getCurrentKey()+"最终结果为：" + reduceState.get());

                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
